package com.baijia.processor;

import com.baijia.dao.PassportDao;
import com.baijia.dao.po.Passport;
import com.baijia.databus.AbstractProcessor;
import com.baijia.databus.ChangedRow;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

/**
 * 监控user表数据变化 Created by dudu on 16/6/21.
 */
@Component
public class UserTableProcessor extends AbstractProcessor {

    private static final Logger logger = LoggerFactory.getLogger(UserTableProcessor.class);

    @Resource
    private PassportDao passportDao;

    @Override
    protected boolean process(List<ChangedRow> changedRows) {
        if (changedRows != null && changedRows.size() > 0) {
            for (ChangedRow columnChange : changedRows) {
                switch (columnChange.getChangedType()) {
                    case UPDATE:
                        doUpdateUser(columnChange);
                        break;
                    case DELETE:
                        break;
                    case INSERT:
                        doInsertUser(columnChange);
                        break;
                    default:
                        break;
                }
            }
        }

        // FIXME 根据操作结果做出返回
        return true;
    }

    private void doInsertUser(ChangedRow columnChange) {

        Map<String, String> newRecord = columnChange.getNewValue();
        logger.info("user id = {}", newRecord.get("ID"));
        logger.debug("[Insert][before change]data = {}", columnChange.getOldValue());
        logger.debug("[Insert][after  change]data = {}", newRecord);

        Passport passport = passportDao.findByMobile(newRecord.get("MOBILE"));
        if (passport != null) {
            logger.warn("user mobile {} is already exist in Passport!", newRecord.get("MOBILE"));
            return;
        }
        passport = new Passport();
        passport.setMobile(newRecord.get("MOBILE"));
        passport.setPassword(newRecord.get("PASSWORD"));
        passport.setUsername(newRecord.get("NAME"));
        passport.setEmail(newRecord.get("EMAIL"));
        passport.setUserId(Long.valueOf(newRecord.get("ID")));
        passportDao.savePassport(passport);
    }

    private void doUpdateUser(ChangedRow columnChange) {

    }

    @Override
    public String topic() {
        return "cdb.user";
    }
}
